![Kinesis FirehoseのS3連携をKinesis StreamとLambdaで実装する #reinvent](https://devio2023-media.developers.io/wp-content/uploads/2014/07/kinesis-icon-gray.png)
Kinesis FirehoseのS3連携をKinesis StreamとLambdaで実装する #reinvent
re:invent 2015 で発表された Kinesis Firehose を使うと、管理画面の操作だけで Kinesis のデータを S3/Redshift に連携します。 Kiensis Firehose を使うとサーバーレスになる上、Kinesis Stream のシャードやイテレーターなどの面倒な操作から開発者を解放してくれます。
ただ残念なことに
- US East (N. Virginia)
- US West (Oregon)
- EU (Ireland)
のリージョンにしか対応しておらず、東京リージョンではまだ使えません。
仕方がないので Kinesis Stream と Lambda で同等の機能を実装してみました。
Kinesis Firehose の S3 連携について
Kinesis Firehose の設定画面は下図のようになっています。
- 連携先のバケット名(S3 bucket)
- S3のキーのプリフィックス(S3 prefix)
を指定すると、S3 バケットに次のように連携されます。
. └── S3Prefix2015 └── 10 └── 16 ├── 22 │ ├── StreamName-1-2015-10-16-22-52-25-ea7c6625-b9d1-413b-ac23-563825c0f8b7 │ └── StreamName-1-2015-10-16-22-57-26-1f89f472-28da-47ac-a6a5-362e0250e6db └── 23 └── StreamName-4-2015-10-16-23-02-27-ed8fc4d7-f7cd-42e4-8417-dca6602f0b70
S3 オブジェクトのキー全体は
S3Prefix2015/10/16/22/StreamName-1-2015-10-16-22-52-25-ea7c6625-b9d1-413b-ac23-563825c0f8b7
というようになっており、
{S3 Prefix}YYYY/MM/DD/HH/{Firehose Stream Name}-{Shard Id?}-YYYY-MM-DD-HH-MM-SS-FF-{Hash}
というような形をしています。
- Shard Id? と思しき箇所は正確には何が使われているのかは不明です。
- Hash で何が使われているのかは不明です。
Lambda の Kinesis -> S3 連携について
実装方針
今回は Firehose の S3 連携処理を、データソースを Kinesis Stream にして Lambda を使って実現します。
- 連携先S3バケット名
- S3 Prefix
- Firehose Stream Name
- Shard Id
は Lambda 関数内で決め打ちし、Hash 関数には uuid v4 を採用します。
Lambda 関数について
Lambda のランタイムには re:invent 2015 ぽさを醸し出すために Python を使います。
Lambda 関数に Kinesis Stream のブループリントがすでに存在するため
- S3 キーの生成
- S3 への Put
を追加したのが以下です。
# vim: set fileencoding=utf8 : import base64 import datetime import uuid import boto3 S3_BUCKET = 'YOUR_BUCKET_NAME' S3_PREFIX = 'prefix_test' STREAM_NAME = 'Stream_Name' SHARD_ID = 1 LINE_TERMINATOR = '\r\n' def get_s3_key(): today = datetime.datetime.utcnow() return "{}{}{}-{}-{}-{}".format( S3_PREFIX, today.strftime('%Y/%m/%d/%H/'), STREAM_NAME, SHARD_ID, today.strftime('%Y-%m-%d-%H-%S-%f')[:-4], uuid.uuid4()) def decode(payload): # TODO # e.g.) string to JSON return payload def process(payload): # TODO # logic comes here return payload def put_to_s3(body): s3_client = boto3.client('s3') s3_client.put_object(Body=body, Bucket=S3_BUCKET, Key=get_s3_key()) def lambda_handler(event, context): buff = [] for record in event['Records']: #Kinesis data is base64 encoded so decode here payload=base64.b64decode(record["kinesis"]["data"]) print "Decoded payload: " + payload buff.append(process(decode(payload))) put_to_s3(LINE_TERMINATOR.join(buff)) return 'Successfully processed {} records.'.format(len(event['Records']))
Kinesis Firehose と同じく、Kinesis のデータを加工せずに S3 に PutObject しているだけですが
- 文字列を他のフォーマット(JSONなど)に変換する関数(decode)
- データ処理する関数(process)
も用意しています。
このLambda関数の登録方法はドキュメントを参照ください。
利用しているモジュールのうち boto3 だけは標準ライブラリではありません。しかし、boto3 は Lambda の Python ランタイムにはインストール済みのため、boto3 モジュールの登録は不要です。
S3 連携を確認
Kinesis Stream にレコード追加
CLI を利用して Kinesis Stream にレコード追加します。
$ aws kinesis put-record --stream-name StreamName --partition-key 1 --data 1,2,3 { "ShardId": "shardId-000000000000", "SequenceNumber": "49555444218629747788391510344658619821542396257728724994" } $ aws kinesis put-record --stream-name StreamName --partition-key 2 --data 'hello world!' { "ShardId": "shardId-000000000000", "SequenceNumber": "49555444218629747788391510344659828747362011642817675266" }
S3 バケットの確認
連携先 S3 バケットをローカルに持ってきて中身を確認します。
$ aws s3 sync s3://BUCKET_NAME BUCKET_NAME $ tree BUCKET_NAME BUCKET_NAME └── prefix_test2015 └── 10 └── 17 └── 00 ├── StreamName-1-2015-10-17-00-01-99-bb8faeb8-8ae5-4e54-87b1-b0d00c773f5b └── StreamName-1-2015-10-17-00-11-67-50b7087e-6e57-40bc-a99d-a4e2277e70b2 4 directories, 2 files
ファイルの中身も確認してみましょう。
$ cat BUCKET_NAME/prefix_test2015/10/17/00/StreamName-1-2015-10-17-00-01-99-bb8faeb8-8ae5-4e54-87b1-b0d00c773f5b 1,2,3 $ cat BUCKET_NAME/prefix_test2015/10/17/00/StreamName-1-2015-10-17-00-11-67-50b7087e-6e57-40bc-a99d-a4e2277e70b2 hello world!
Kinesis に Put したデータが無事 S3 に連携されていますね。
まとめ
この程度の処理であれば Lambda 関数を書くのもそれほど手間ではありませんが、さっさと Kinesis Firehose が東京リージョンにやってきて、画面ポチだけで S3 連携されるようになって欲しいですね!